import org.apache.spark.sql.SparkSession
import org.junit.Test

class 连接操作 {
  val spark = SparkSession.builder()
    .master("local[6]")
    .appName("Test")
    .getOrCreate()

  import spark.implicits._
  import org.apache.spark.sql.functions._

  private val person = Seq((0, "Lucy", 0), (1, "Lily", 0), (2, "Tim", 2), (3, "Danial", 0)).toDF("id", "name", "cityID")
  person.createOrReplaceTempView("person")
  private val cities = Seq((0, "Beijing"), (1, "Shanghai"), (2, "Guangzhou")).toDF("id", "name")
  cities.createOrReplaceTempView("cities")


  /**
    * 简单连接的案列
    */
  @Test
  def simplejoin(): Unit = {
    val df = person.join(cities, person.col("cityID") === cities.col("id"))
      .select(person.col("id"), person.col("name"), cities.col("name") as "city")
    df.createOrReplaceTempView("user_city")
    spark.sql("select id,name,city from user_city where city = 'Beijing'").show()
  }

  /**
    * 笛卡尔积,交叉
    */
  @Test
  def crossJoin(): Unit = {
    person.crossJoin(cities)
      .where(person.col("cityID") === cities.col("id"))
      .show()
    spark.sql(" select u.id,u.name,c,name from person u cross join cities c where u.cityID = c.id")
      .show()
  }

  /**
    * 内连接
    */
  @Test
  def innerJoin(): Unit ={
    person.join(cities,person.col("cityID") === cities.col("id"),joinType = "inner")
      .show()
    spark.sql("select p.id,p.name,c.name " +
      "from person p join cities c on p.cityID = c.id").show()
  }

  /**
    * 全外连接
    */
  @Test
  def fullJoin(): Unit ={
     person.join(cities,person.col("cityID") === cities.col("id"),joinType = "outer").show()
    spark.sql("select p.id,p.name,c.name " +
      "from person p full outer join cities c on p.cityID = c.id").show()
  }

  /**
    * left anti只显示左侧没有连接上的,不显示右侧数据
    */
  @Test
  def leftanti(): Unit ={
    person.join(cities,person.col("cityID") === cities.col("id"),joinType = "leftanti").show()
    spark.sql("select p.id,p.name,c.name " +
      "from person p left anti join cities c on p.cityID = c.id").show()
  }

  /**
    * left semi只显示左侧连接上的,不显示右侧数据
    */
  @Test
  def leftsemi(): Unit ={
    person.join(cities,person.col("cityID") === cities.col("id"),joinType = "leftsemi").show()
    spark.sql("select p.id,p.name,c.name " +
      "from person p left semi join cities c on p.cityID = c.id").show()
  }
}
